package org.nbict.iot.task.realtime.func;

import com.alibaba.fastjson.JSONObject;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by songseven on 18/6/26.
 */
public class GisResolver extends BaseFunction {

    private static Logger log = LoggerFactory.getLogger(GisResolver.class);
    //    private static final String[] tags = {"01", "05", "07"}; //整车、定位、报警
    private static final Map<String, String[]> tags = new HashMap<>();

    static {
        tags.put("01", new String[]{"status", "charged_state"});
        tags.put("05", new String[]{"lon", "lat"});
        tags.put("07", new String[]{"alarm_level"});
    }

    // timestamp
    // 1: car_status charge_status
    // 5: lon lat
    // 7: alarm_status
    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
        JSONObject upload = (JSONObject) tuple.get(3); //info

        Values values = new Values();

//        Arrays.asList(tags).forEach(t -> {
//            JSONObject info = (JSONObject) upload.get(t);
//            if (info != null) {
//                Arrays.asList(tags.get(t)).forEach(f -> {
//                    values.add(info.get(f));
//                });
//            }
//        });
        tags.keySet().forEach(k -> {
            JSONObject info = (JSONObject) upload.get(k);
            if (info != null) {
                Arrays.asList(tags.get(k)).forEach(f -> {
                    values.add(info.get(f));
                });
            } else {
                String [] fields = tags.get(k);
                Arrays.asList(fields).forEach(s -> {
                    values.add(null);
                });
            }
        });

        //log.info("values=>" + values.toString());
        collector.emit(values);
    }

//    public static void main(String[] args) {
//        System.out.println(new GisResolver().tags);
//    }
}
